package integration.data.task.api.runner;

import integration.data.task.writer.Writer;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.collector.TaskPluginCollector;
import io.github.bigbird0101.datatransfer.record.RecordReceiver;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class WriterRunner extends AbstractRunner implements Runnable {

    private RecordReceiver recordReceiver;

    private Writer writer;

    public WriterRunner(Writer writer) {
        this.writer = writer;
    }

    public void setRecordReceiver(RecordReceiver recordReceiver) {
        this.recordReceiver = recordReceiver;
    }

    public void setJobConf(String jobConf) {
        super.setJobConf(jobConf);
        writer.setConfig(jobConf);
    }

    @Override
    public void setTaskId(String taskId) {
        super.setTaskId(taskId);
        // writer taskID
        writer.setTaskId(taskId);
    }

    @Override
    public void setTaskPluginCollector(TaskPluginCollector pluginCollector) {
        writer.setTaskPluginCollector(pluginCollector);
    }

    @Override
    public void shutdown() {
        recordReceiver.shutdown();
    }

    @Override
    public void destroy() {
        writer.destroy();
    }

    @Override
    public void run() {
        try {
            writer.init();
            writer.prepare();
            writer.startWrite(recordReceiver);
            super.markSuccess();
        } catch (Throwable e) {
            LogUtils.error(log, "Writer Runner Received Exceptions:", e);
            super.markFail(e);
        } finally {
            log.debug("task writer starts to do destroy ...");
            this.destroy();
        }
    }
}
